package com.aiwiown.snackmq.broker.handler;

import com.aiwiown.snackmq.broker.auth.AuthorizationService;
import com.aiwiown.snackmq.cluster.raft.closeure.SnackMQApplyClosure;
import com.aiwiown.snackmq.cluster.raft.entity.SnackMQRaftLogEntry;
import com.aiwiown.snackmq.cluster.raft.manager.SnackMQRaftNodeManager;
import com.aiwiown.snackmq.common.auth.Action;
import com.aiwiown.snackmq.common.exception.AuthorizationException;
import com.aiwiown.snackmq.common.handler.MessageHandler;
import com.aiwiown.snackmq.common.handler.RequestContext;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.common.metrics.MessageMetricEmitter;
import com.aiwiown.snackmq.common.protocol.OperationType;
import lombok.extern.slf4j.Slf4j;

/**
 * 【重构】处理消费者提交位点 (COMMIT_OFFSET) 请求的处理器。
 * (此类原名为 AckMessageHandler，建议重命名为 CommitOffsetHandler 以更准确地反映其功能)
 * <p>
 * 此版本增加了完整的 Leader 检查、重定向和权限验证逻辑，以确保在集群环境中的健壮性和安全性。
 */
@Slf4j
public class AckMessageHandler implements MessageHandler {

    private final SnackMQRaftNodeManager dataRaftNodeManager;
    private final AuthorizationService authorizationService;
    private final MessageMetricEmitter metricEmitter;

    /**
     * 构造函数，初始化处理器。
     *
     * @param dataRaftNodeManager  Raft 节点管理器。
     * @param authorizationService 授权服务。
     * @param metricEmitter        指标发射器。
     */
    public AckMessageHandler(SnackMQRaftNodeManager dataRaftNodeManager,
                             AuthorizationService authorizationService,
                             MessageMetricEmitter metricEmitter) {
        this.dataRaftNodeManager = dataRaftNodeManager;
        this.authorizationService = authorizationService;
        this.metricEmitter = metricEmitter;
    }

    /**
     * 处理 COMMIT_OFFSET 消息请求。
     * <p>
     * 1. 检查当前节点是否为数据 Raft 组的 Leader，如果不是则重定向客户端。
     * 2. 对请求进行权限验证，确保客户端有权消费该 Topic 和 ConsumerGroup。
     * 3. 参数校验通过后，将一个干净的位点信息提交到 Raft 日志。
     * 4. 统一处理所有异常并向客户端返回明确的错误信息。
     *
     * @param ctx           请求上下文。
     * @param clientRequest 客户端请求消息。
     */
    @Override
    public void handle(RequestContext ctx, Message clientRequest) {
        try {
            // 1. 检查 Leader 状态，如果不是 Leader 则重定向
            if (!dataRaftNodeManager.isLeader()) {
                String leaderAddress = dataRaftNodeManager.getLeaderAddress();
                if (leaderAddress != null) {
                    log.debug("Current node is not the data leader. Redirecting commit offset request to {}", leaderAddress);
                    Message redirectResponse = Message.builder()
                            .type(MessageType.RESPONSE)
                            .messageId(clientRequest.getMessageId())
                            .status(MessageStatus.REDIRECT)
                            .property("redirectAddress", leaderAddress)
                            .build();
                    ctx.writeResponse(redirectResponse);
                    return;
                }
                log.warn("Current node is not leader, but could not determine the leader's address. Rejecting request.");
                // 【优化】加入 groupId，使错误信息更具诊断性
                String errorMessage = String.format("Service unavailable: No data leader elected yet for group [%s]. Please retry.", dataRaftNodeManager.getGroupId());
                SnackMQApplyClosure.sendErrorResponse(ctx, clientRequest, errorMessage);
            }

            // 2. 从客户端请求中提取所有必要信息
            String consumerGroup = clientRequest.getProperty("consumerGroup");
            String topic = clientRequest.getTopic();
            String partitionId = clientRequest.getProperty("partition");
            String offset = clientRequest.getProperty("offset");

            // 3. 参数校验
            if (consumerGroup == null || topic == null || partitionId == null || offset == null) {
                log.error("Invalid COMMIT_OFFSET request, missing required properties: {}", clientRequest);
                SnackMQApplyClosure.sendErrorResponse(ctx, clientRequest, "Invalid request, missing properties.");
                return;
            }

            // 4. 授权检查
            authorizationService.authorize(clientRequest.getProperty("authToken"), Action.CONSUME, new AuthorizationService.Resource("TOPIC", topic));
            authorizationService.authorize(clientRequest.getProperty("authToken"), Action.CONSUME, new AuthorizationService.Resource("CONSUMER_GROUP", consumerGroup));

            // 5. 创建一个干净的 Message 对象用于写入 Raft 日志
            Message logMessage = Message.builder()
                    .topic(topic)
                    .property("consumerGroup", consumerGroup)
                    .property("partition", partitionId)
                    .property("offset", offset)
                    .build();

            // 6. 创建 Raft 日志条目并提议
            SnackMQRaftLogEntry logEntry = new SnackMQRaftLogEntry(OperationType.COMMIT_OFFSET, logMessage);
            SnackMQApplyClosure closure = new SnackMQApplyClosure(clientRequest, ctx, this.metricEmitter);
            dataRaftNodeManager.apply(logEntry, closure);

        } catch (AuthorizationException e) {
            log.warn("Authorization failed for commit offset request: {}", e.getMessage());
            SnackMQApplyClosure.sendErrorResponse(ctx, clientRequest, "Permission denied: " + e.getMessage());
        } catch (Exception e) {
            log.error("Error processing commit offset request: {}", clientRequest, e);
            SnackMQApplyClosure.sendErrorResponse(ctx, clientRequest, "Internal server error during offset commit.");
        }
    }

    /**
     * 获取该处理器处理的消息类型。
     *
     * @return 消息类型为 COMMIT_OFFSET。
     */
    @Override
    public MessageType getMessageType() {
        return MessageType.COMMIT_OFFSET;
    }
}